package org.ala.tcp_feign_client.context.reconnect;

import java.util.HashSet;
import java.util.LinkedList;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;
import java.util.function.Consumer;

import org.ala.tcp_feign_client.ITcpFeignClient;
import org.ala.tcp_feign_common.logs.LoggerFactory;
import org.slf4j.Logger;


/**
 * 重连定时任务
 * <p>	对于失败或断掉的连接，需要重连机制
 *
 * @author 骆毅(ala)
 * @date 2024年4月22日
 */
public class ReconnectScheduled {

	
	static Logger log = LoggerFactory.getLogger("tcp-feign", "client");
	
	
	/**
	 * 定时任务
	 */
	protected ScheduledExecutorService schedule;
	/**
	 * 重连间隔
	 * <p>	单位：毫秒
	 * <p>	默认：5s
	 */
	protected long interval = 5000l;
	/**
	 * 重连任务列表
	 */
	protected Set<ReConnectTask> tasks = new HashSet<>();
	/**
	 * 重连任务执行过程中如果出现新任务，吃到slaver队列里
	 * <p>	重连任务迭代一轮完成后，slaver中的任务还给tasks
	 */
	protected Set<ReConnectTask> slaver = new HashSet<>();
	/**
	 * 是否在迭代
	 */
	protected volatile boolean isIterator;
	
	
	/**
	 * 初始化
	 */
	public void start() {
		schedule = Executors.newScheduledThreadPool(1);
		schedule.scheduleAtFixedRate(() -> reconnect(), interval, interval, TimeUnit.MILLISECONDS);
	}
	/**
	 * 重连
	 */
	protected void reconnect() {
//		log.info("[ReconnectScheduled reconnect] 开始重连任务");
		isIterator = true;
		try {
			LinkedList<ReConnectTask> succList = new LinkedList<>();
			for (ReConnectTask task : tasks) {
				//	如果已经链接成功，无需再次重连
				if (task.client.isActive()) {
					succList.add(task);
					continue;
				}
				
				//	尝试重连
				try {
					task.client.close();
					task.client.connect();
					succList.add(task);
					
					if (task.consumer != null) {
						task.consumer.accept(task.client);
					}
				} catch (Exception e) {
					log.warn("[ReconnectScheduled reconnect] 重连失败，等待下次重试", e);
				}
			}
			while (!succList.isEmpty()) {
				tasks.remove(succList.poll());
			}
		} finally {
			isIterator = false;
			for (ReConnectTask task : slaver) {
				tasks.add(task);
				slaver.remove(task);
			}
		}
	}
	/**
	 * 追加重连任务
	 */
	public void appendReconnect(ITcpFeignClient client, Consumer<ITcpFeignClient> consumer) {
		ReConnectTask task = new ReConnectTask(client, consumer);
		//	检测是否已经存在
		if (tasks.contains(task) || slaver.contains(task)) {return;}
		
		if (isIterator) {slaver.add(task);}
		else {tasks.add(task);}
	}
	/**
	 * 检测链接是否在重连队列中
	 */
	public boolean contains(ITcpFeignClient client) {
		ReConnectTask task = new ReConnectTask(client, null);
		return tasks.contains(task) || slaver.contains(task);
	}
	
	
	/**
	 * 等待重连任务
	 */
	public static class ReConnectTask {
		ITcpFeignClient client;
		Consumer<ITcpFeignClient> consumer;
		public ReConnectTask(ITcpFeignClient client, Consumer<ITcpFeignClient> consumer) {
			super();
			this.client = client;
			this.consumer = consumer;
		}
		@Override
		public int hashCode() {
			return client.host().hashCode() + client.port() + client.suffix().hashCode();
		}
		@Override
		public boolean equals(Object obj) {
			if (!(obj instanceof ReConnectTask)) {return false;}
			ReConnectTask task = (ReConnectTask) obj;
			return task.client.host().equals(client.host()) 
					&& task.client.port() == client.port() 
					&& task.client.suffix().equals(client.suffix());
		}
		
	}
}
